#include "TimerManager.h"
#include "eventloop/Eventloop.h"
#include "eventloop/IOchannel.h"
#include "log/Logger.h"
#include "thread/current_thread.h"
#include "utility/assertion.h"
#include "utility/cast.h"
#include "utility/feature.h"
#include "wrap/timerfd.h"
#include "wrap/unistd.h"

#include <atomic>

namespace _ {

static TimerID next_ID() {
    static std::atomic<TimerID> totalTimerID = TIMER_ID_MAX;
    ++totalTimerID;
    if (totalTimerID == UINT64_MAX) {
        ++totalTimerID;
    }
    return totalTimerID;
}

static int create_timerfd() {
    const auto fd = Wrap::timerfd_create_(CLOCK_MONOTONIC, Wrap::tfd_nonblock() | Wrap::tfd_cloexec());
    if (fd < 0) {
        LOG_SYSFATAL << "Failed";
    }
    return fd;
}

static void read_timerfd(int fd, Utility::timepoint t) {
    using ReadType = uint64_t;
    ReadType buffer = 0;
    const auto n = Wrap::read_(fd, &buffer, sizeof(ReadType));
    LOG_VERBOSE << "timer timeout " << buffer << " at " << Utility::to_string(t);
    if (n == sizeof(ReadType)) {
        return;
    }
    LOG_ERROR << "timer read_ " << n << " bytes instead of " << sizeof(ReadType); 
}

static void reset_timerfd(int fd, Utility::timepoint t) {
    static const auto DELAY_IN_COUNT = Utility::count(Utility::cast_duration(100e-6));
    const auto count = std::max(Utility::count(t - Utility::now()), DELAY_IN_COUNT);
    struct itimerspec itNew {};
    Cast::assign(itNew.it_value.tv_sec, count/Utility::COUNT_PER_SEC);
    Cast::assign(itNew.it_value.tv_nsec, count % Utility::COUNT_PER_SEC);
    if (Wrap::timerfd_settime_(fd, 0, &itNew, nullptr) != 0) {
        LOG_SYSERROR << "timerfd_settime_";
    }
}

}

TimerManager::TimerManager(Eventloop &loop):
    _timers{},
    _activeTimerIDMap{},
    _cancelingTimerIDs{},
    _loop{loop},
    _channel{std::make_unique<IOchannel>(
            loop, _::create_timerfd(), std::string(__func__) + CurrentThread::tid_string())},
    _callingExpiredTimers{false}
{
    _channel->set_read_callback(
        [this](Utility::timepoint /*t*/){ _read_callback(); });
    _channel->enable_reading();
}

TimerManager::~TimerManager() {
    _channel->disable();
    _channel->remove();
    Wrap::close_(_channel->fd());
    _timers.clear();
}

TimerID TimerManager::add(const TrivialFunc &callback, Utility::duration delay, bool isOneShot) {
    assert(callback);
    const auto id = _::next_ID();
    _loop.run_in_loop([this, callback, expiration = Utility::now() + delay, 
                       interval = isOneShot ? Utility::duration::zero() : delay, id]{
        _loop.ASSERT_IN_THIS_THREAD();
        if (!_insert(Timer{callback, expiration, interval, id})) {
            return;
        }
        _::reset_timerfd(_channel->fd(), expiration);
    });
    return id;
}

void TimerManager::cancel(TimerID id) {
    _loop.run_in_loop([this, id](){
        _loop.ASSERT_IN_THIS_THREAD();
        assert(_timers.size() == _activeTimerIDMap.size());
        const auto iter = _activeTimerIDMap.find(id);
        if (iter != _activeTimerIDMap.cend()) {
            MAYBE_UNUSED const auto n = _timers.erase({iter->second, iter->first});
            assert(n == 1);
            _activeTimerIDMap.erase(iter);
        }
        else if (_callingExpiredTimers) {
            _cancelingTimerIDs.insert(id);
        }
        assert(_timers.size() == _activeTimerIDMap.size());
    });
}

bool TimerManager::_insert(Timer &&timer) {
    _loop.ASSERT_IN_THIS_THREAD();
    assert(_timers.size() == _activeTimerIDMap.size());
    const auto expiration = timer.expiration();
    const auto sequence = timer.id();
    MAYBE_UNUSED const auto result1 = _activeTimerIDMap.insert({sequence, expiration}).second;
    assert(result1);
    const auto isFirst = _timers.empty() || expiration < _timers.cbegin()->first.first;
    MAYBE_UNUSED const auto result2 = _timers.insert({{expiration, sequence}, std::move(timer)}).second;
    assert(result2);
    assert(_timers.size() == _activeTimerIDMap.size());
    return isFirst;
}

void TimerManager::_read_callback() {
    _loop.ASSERT_IN_THIS_THREAD();
    const auto now = Utility::now();
    _::read_timerfd(_channel->fd(), now);
    _run_expired_funcs(_take_expired_timers(now), now);
    const auto nextExpire = _timers.empty() ? Utility::timepoint::min() 
                                           : _timers.cbegin()->first.first;
    if (!Utility::is_valid(nextExpire)) {
        return;
    }
    _::reset_timerfd(_channel->fd(), nextExpire);
}

std::vector<Timer> TimerManager::_take_expired_timers(Utility::timepoint now) {
    assert(_timers.size() == _activeTimerIDMap.size());
    const auto expiredEnd = _timers.lower_bound({now, TIMER_ID_MAX});
    assert(expiredEnd == _timers.cend() || now < expiredEnd->first.first);
    std::vector<Timer> expiredTimers{};
    for (auto x = _timers.begin(); x != expiredEnd; ++x) {
        MAYBE_UNUSED const auto n = _activeTimerIDMap.erase(x->first.second);
        assert(n == 1);
        expiredTimers.emplace_back(std::move(x->second));
    }
    _timers.erase(_timers.cbegin(), expiredEnd);
    assert(_timers.size() == _activeTimerIDMap.size());
    return expiredTimers;
}

void TimerManager::_run_expired_funcs(
    std::vector<Timer> &&expiredTimers, Utility::timepoint now) {
    assert(_timers.size() == _activeTimerIDMap.size());
    _callingExpiredTimers = true;
    _cancelingTimerIDs.clear();
    for (const auto &x : expiredTimers) {
        x.run();
    }
    _callingExpiredTimers = false;
    for (auto &x : expiredTimers) {
        if (!x.is_repeat() || _cancelingTimerIDs.find(x.id()) != _cancelingTimerIDs.cend()) {
            continue;
        }
        x.restart(now);
        _insert(std::move(x));
    }
}
